package middleware.kafka.init_kafka;

import middleware.kafka.Serializer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Future;

/**
 * 生产者客户端示例代码
 *
 * @Author suxiaoxiao
 * @Date 2019/6/26 0:42
 * @Version 1.0
 */
public class KafkaProducerAnalysis {

    public static final String brokerList = "47.104.178.113:9092";
    public static final String topic = "topic-demo";

    public static Properties initConfig() {
        Properties props = new Properties();
        /**
         * 该参数用来指定生产者客户端连接 Kafka 集群所需的 broker 地址清单，
         * 具体的内容格式为 host1:port1,host2:port2
         */
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerList);
        /**
         * key 序列化器
         */
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        /**
         * value 序列化器
         */
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                StringSerializer.class.getName());
        /**
         * 分区器
         */
        props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, DemoPartitioner.class.getName());
        /**
         * 拦截器
         */
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
                ProducerInterceptorPrefixPlus.class.getName() + "," + ProducerInterceptorPrefix.class.getName());
        /**
         * 客户端 id
         */
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "producer.client.id.demo");
        /**
         * 重试 10 次，
         * 如果重试10次之后还没有恢复，那么会抛出异常。
         */
        props.put(ProducerConfig.RETRIES_CONFIG, 10);
        return props;
    }

    public static void main(String[] args) {

        Properties props = initConfig();
        /**
         * 创建一个生产者实例 KafkaProducer
         *
         * KafkaProducer 是线程安全的
         */
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

//        Company company = Company.builder().name("hiddenkafka").address("China").build();
        /**
         * 消息对象
         *
         *  private final String topic; // 主题
         *  private final Integer partition; // 分区号
         *  private final Headers headers; // 消息头部
         *  private final K key; // 键
         *  private final V value; // 值
         *  private final Long timestamp; // 消息时间戳
         *
         */
        ProducerRecord<String, String> record = new ProducerRecord<>(
                topic, 0, "KafkaKey", "Kafka");
        try {
            /**
             * 发后即忘（fire-and-forget）
             * 只发送不管消息是否到达
             */
            producer.send(record);
            /**
             * 同步发送的两种方法
             * 1. 调用 get() 方法
             */
//            producer.send(record).get();
            /**
             * -------------------------------
             * 2. 使用 Future
             * 这样可以获取一个 RecordMetadata 对象，
             * 在 RecordMetadata 对象里包含了消息的一些元数据信息，
             * 比如当前消息的主题、分区号、分区中的偏移量（offset）、时间戳等。
             * 如果在应用代码中需要这些信息，则可以使用这个方式。
             */
//            Future<RecordMetadata> future = producer.send(record);
//            RecordMetadata metadata = future.get();
//            System.out.println(metadata.topic() + "-" +  metadata.partition() + ":" + metadata.offset());
            /**
             * ---------------------------------
             * 异步回调
             */
//            producer.send(record, (metadata, exception) -> {
//                if (exception != null) {
//                    exception.printStackTrace();
//                } else {
//                    System.err.println(metadata.topic() + "-" +
//                            metadata.partition() + ":" + metadata.offset());
//                }
//            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }

}
